package javaVersion.learn.batchProcess;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

/**
 * Java版本的Flink WordCount
 */
public class Pro4_WordCount {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // TODO: 2020/7/16 创建数据源
        ArrayList<String> list = new ArrayList<>();
        list.add("a");
        list.add("a");
        list.add("a");
        list.add("bb");
        list.add("bb");
        list.add("c");
        DataSource<String> source = env.fromCollection(list);

        // TODO: 2020/7/16 使用数据源、
        //使用java普通方法编写
//        standMethod(source);
        //使用lambda表达式编写
        lambdaMethod(source);

    }

    private static void lambdaMethod(DataSource<String> source) throws Exception {
        //对数据中每个项目进行操作 FlatMap
        source.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> out) throws Exception {
                out.collect(new Tuple2<>(s,1));
            }
        })
                .groupBy(0)
                .reduce((t2, t1) -> new Tuple2<>(t1.f0,t1.f1+t2.f1))
                .print();
    }

    private static void standMethod(DataSource<String> source) throws Exception {
//        //对数据中每个项目进行操作 FlatMap
//        source.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
//            @Override
//            public void flatMap(String s, Collector<Tuple2<String, Integer>> out) throws Exception {
//                out.collect(new Tuple2<>(s,1));
//            }
//        })
//                .groupBy(0)
//                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
//                    @Override
//                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> t2, Tuple2<String, Integer> t1) throws Exception {
//                        return new Tuple2<>(t1.f0,t1.f1+t2.f1);
//                    }
//                })
//                .print();
    }
}
